package own.stu.jobgib.playown.alg.consistentHashing;

import lombok.Data;
import lombok.extern.slf4j.Slf4j;

import java.util.HashMap;
import java.util.LinkedList;
import java.util.Map;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.RecursiveAction;
import java.util.concurrent.atomic.AtomicInteger;
//import org.apache.commons.cli.CommandLine;
//import org.apache.commons.cli.CommandLineParser;
//import org.apache.commons.cli.DefaultParser;
//import org.apache.commons.cli.HelpFormatter;
//import org.apache.commons.cli.Options;
//import org.apache.commons.logging.Log;
//import org.apache.commons.logging.LogFactory;

@Slf4j
public class Test extends RecursiveAction {
    private static final long serialVersionUID = -8032915917030559798L;

    private int start;
    private int finish;
    private Map<Node<String>, AtomicInteger> status;
    private ConsistentHashing<String> consistentHashing = null;
    private int threshold = 1000000;

    // Average pixels from source, write results into destination.
    protected void computeDirectly() {
        for (int index = start; index <= finish; index++) {
            String key = "key" + index;
            Node<String> keyToHostNode = consistentHashing.getNodeByKey(key);
            AtomicInteger i = status.get(keyToHostNode);
            i.addAndGet(1);
        } //end for i 
    }

    @Override
    protected void compute() {
        if (finish - start <= threshold) {
            log.info(Thread.currentThread() + ":" + start + ":" + finish);
            computeDirectly();
            return;
        }

        int split = (start + finish) / 2;
        Test left = new Test();
        left.setStart(start);
        left.setFinish(split);
        left.setShared(consistentHashing);
        left.setStatus(status);

        Test right = new Test();
        right.setStart(split + 1);
        right.setFinish(finish);
        right.setShared(consistentHashing);
        right.setStatus(status);

        invokeAll(left, right);
    }

    // Plumbing follows.
    public static void main(String[] args) throws Exception {
        LinkedList<Node<String>> nodeList = new LinkedList<Node<String>>();
        Map<Node<String>, AtomicInteger> status = new HashMap<>();
        int hostSize = 100;
        //int finish = Integer.MAX_VALUE;
        int finish = 1000000000;
        int begin = 0;
        int threshold = finish / 100;

        /*Options options = new Options();
        options.addOption("begin", "begin", true, "开始key的大小");
        options.addOption("finish", "finish", true, "结束key的大小");
        options.addOption("hostSize", "hostSize", true, "服务器个数");
        options.addOption("threshold", "threshold", true, "fork最小处理块行数");

        CommandLineParser parser = new DefaultParser();
        CommandLine cmdline = parser.parse(options, args);

        if (args == null || args.length == 0) {
            // 打印命令行定义
            HelpFormatter formatter = new HelpFormatter();
            formatter.printHelp("product.distributed.consistenthashing", options);
        }

        if (cmdline.hasOption("begin")) {
            begin = Integer.parseInt(cmdline.getOptionValue("begin"));
        }
        if (cmdline.hasOption("finish")) {
            finish = Integer.parseInt(cmdline.getOptionValue("finish"));
        }
        if (cmdline.hasOption("hostSize")) {
            hostSize = Integer.parseInt(cmdline.getOptionValue("hostSize"));
        }
        if (cmdline.hasOption("threshold")) {
            threshold = Integer.parseInt(cmdline.getOptionValue("threshold"));
        }*/

        ConsistentHashing<String> chash = new ConsistentHashing<String>();
        for (int j = 0; j < hostSize; j++) {
            MockNode node = new MockNode();
            node.setName("host" + j);
            node.setHost("10.10.127." + j);
            node.setPort(6789);
            node.setWeight(10);
            status.put(node, new AtomicInteger(0));
            nodeList.add(node);
            chash.addHostNode(node);
        } //end for j

        log.info("getNodeSize:" + chash.getNodeSize());
        log.info("getVirtualNodeSize:" + chash.getVirtualNodeSize());

        ForkJoinPool pool = new ForkJoinPool();
        Test forkhash = new Test();
        forkhash.setStart(begin);
        forkhash.setFinish(finish);
        forkhash.setShared(chash);
        forkhash.setStatus(status);
        forkhash.setThreshold(threshold);

        pool.invoke(forkhash);
        long sum = 0;
        for (Node<String> index : nodeList) {
            sum += status.get(index)
                    .get();
            log.info("index:" + index + ":" + status.get(index));
        } //end for nodeList
        log.info("sum:" + sum);
    }

    public int getStart() {
        return start;
    }

    public void setStart(int start) {
        this.start = start;
    }

    public int getFinish() {
        return finish;
    }

    public void setFinish(int finish) {
        this.finish = finish;
    }

    public Map<Node<String>, AtomicInteger> getStatus() {
        return status;
    }

    public void setStatus(Map<Node<String>, AtomicInteger> status) {
        this.status = status;
    }

    public ConsistentHashing<String> getShared() {
        return consistentHashing;
    }

    public void setShared(ConsistentHashing<String> consistentHashing) {
        this.consistentHashing = consistentHashing;
    }

    public int getThreshold() {
        return threshold;
    }

    public void setThreshold(int threshold) {
        this.threshold = threshold;
    }


    /**
     * 此类主要目的:
     * 0、类封装了机器节点的信息 ，如name、password、host、port等
     *
     * @author <a href="mailto:liu_jing_wei@sohu.com">Dreams Liu </a>
     * @version $Revision 1.1 $ 2018年8月10日 上午8:40:44
     */
    @Data
    public static class MockNode implements Node<String> {
        /**
         * node名称
         */
        private String name;

        /**
         * node地址
         */
        private String host;

        /**
         * node端口
         */
        private int port = 0;

        /**
         * 权重（虚拟node的个数）
         */
        private int weight = 1;

        /**
         * node的索引顺序
         * 保留未使用
         */
        private String index = "";

        /**
         * 失败率
         * 保留未使用
         */
        private float failureRate = 0;

        public String getVirtualNodeName(int index) {
            String rv = "";
            rv = this.getName() + "#" + index;
            return rv;
        }

        @Override
        public String getResource() {
            return toString();
        }

        @Override
        public String toString() {
            return this.name + "-" + this.host + ":" + this.port + "#" + this.weight;
        }
    }
}
